package com.atguigu.day02;


import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.sql.Timestamp;
import java.util.Calendar;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * filter的使用
 */
public class Example3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env
                .addSource(new MySource())
                .filter(x->x.name.equals("zhang3"))
                .print();
        env
                .addSource(new MySource())
                .filter(new FilterFunction<Event>() {
                    @Override
                    public boolean filter(Event event) throws Exception {
                        return event.name.equals("zhang3");
                    }
                })
                .print();
        env
                .addSource(new MySource())
                .filter(new MyFilter())
                .print();
        env.execute();
    }

    public static class MyFilter implements FilterFunction<Event>{
        @Override
        public boolean filter(Event event) throws Exception {
            return event.name.equals("zhang3");
        }
    }

    public static class MySource implements SourceFunction<Event> {
        private Boolean flag = true;
        private String[] nameArr = {"zhang3", "li4", "wang5"};
        private String[] urlArr = {"baidu.com", "jd.com", "google.com", "/home"};
        private Random random = new Random();

        @Override
        public void run(SourceContext<Event> sourceContext) throws Exception {
            while (flag) {
                sourceContext.collect(new Event(
                        nameArr[random.nextInt(nameArr.length)],
                        urlArr[random.nextInt(urlArr.length)],
                        Calendar.getInstance().getTimeInMillis()
                ));
                TimeUnit.MILLISECONDS.sleep(500);
            }
        }

        @Override
        public void cancel() {

        }
    }

    public static class Event {
        public String name;
        public String url;
        public Long timestamp;

        public Event(String name, String url, Long timestamp) {
            this.name = name;
            this.url = url;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "Event{" +
                    "name='" + name + '\'' +
                    ", url='" + url + '\'' +
                    ", timestamp=" + new Timestamp(timestamp) +
                    '}';
        }

        public Event() {
        }
    }
}
